package com.ibm.cps.join;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;


public class JoinerFactory {
	private static Map<String, Class<? extends AbstractJoiner>> type2Joiner = new HashMap<String, Class<? extends AbstractJoiner>>();
	private static Logger logger = Logger.getLogger(JoinerFactory.class);
	
	static{
		type2Joiner.put(MessageFields.MONGO, MongoJoiner.class);
	}
	
	public static AbstractJoiner createJoiner(String metadata){
		try {
			ObjectNode node = JSONUtility.fromObjectString(metadata);
			JSONUtility.checkNotNull(node, MessageFields.TYPE);
			String type = JSONUtility.GetString(MessageFields.TYPE, node);
			if (!type2Joiner.containsKey(type)) {
				throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "invalid key in JoinerFactory:[" + type + "]");
			}
			try {
				Constructor<? extends AbstractJoiner> constructor = type2Joiner.get(type).getConstructor(String.class);
				return constructor.newInstance(metadata);
			} catch (NoSuchMethodException | SecurityException | InstantiationException | IllegalAccessException
					| IllegalArgumentException | InvocationTargetException e) {
				e.printStackTrace();
				logger.error(e.toString());
			}
		} catch (CPSException e) {
			e.printStackTrace();
		}
		return null;
	}
	
}
